// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: jshin

#include "coin2/flow/topic_subscription.h"

#include <memory>
#include <set>
#include <string>
#include <utility>

#include "coin2/flow/topic_string.h"
#include "coin2/flow/symbol_group.h"

namespace coin2::exchange::base::feed {

using coin2::flow::SymbolGroupConfig;

void TopicRegister::RegisterLiveShortTopic(
    const MarketExchangeApi& mea,
    executor::TopicRecordSubscriber* subscriber,
    executor::TopicRecordDistributor* dist) {
  // "spot_okex_v3"
  std::string sub(mea.SnakeString());
  dist->Subscribe(sub, subscriber);
}

void TopicRegister::RegisterLiveLongTopic(
  const MarketExchangeApi& mea,
  executor::TopicRecordSubscriber* subscriber,
  executor::TopicRecordDistributor* dist) {
  // "spot_okex_v3_a1"
  // "spot_okex_v3_a2"
  // "spot_okex_v3_b1"
  // "spot_okex_v3_b2"
  // ..
  std::string sub(mea.SnakeString());
  sub += "_a1";
  for (char group = 'a'; group <= 'z'; group++) {
    sub[sub.size() - 2] = group;
    for (char worker_id = '1'; worker_id <= '2'; worker_id++) {
      sub[sub.size() - 1] = worker_id;
      dist->Subscribe(sub, subscriber);
    }
  }
}

void TopicRegister::RegisterLiveLongTopicArb(
  const MarketExchangeApi& mea,
  executor::TopicRecordSubscriber* subscriber,
  executor::TopicRecordDistributor* dist,
  const ArbitrationConfigMap& arbitration_config,
  int32_t num_workers) {
  arbitrators_.clear();

  // "spot_okex_v3_a1"
  // "spot_okex_v3_a2"
  // "spot_okex_v3_b1"
  // "spot_okex_v3_b2"
  // ..
  std::string sub(mea.SnakeString());
  sub += "_a1";
  for (char group = 'a'; group <= 'z'; group++) {
    auto p = arbitrators_.insert(
      std::make_pair(
        group,
        std::make_unique<executor::TopicRecordArbitrator>(
          subscriber, arbitration_config.GetWindowSize(mea, group))));
    sub[sub.size() - 2] = group;
    for (int worker = 1; worker <= num_workers; ++worker) {
      sub[sub.size() - 1] = std::to_string(worker)[0];
      dist->Subscribe(sub, p.first->second.get());
    }
  }
}

std::set<std::string> TopicRegister::RegisterFlowTopicsArb(
    const MarketExchangeApi& mea,
    const std::string& workers,
    const std::vector<std::unique_ptr<symbology::IProduct>>& products_,
    executor::TopicRecordSubscriber* subscriber,
    executor::TopicRecordDistributor* dist,
    const ArbitrationConfigMap& arbitration_config) {
  std::set<std::string> all_topics;
  // interate symbols
  for (const auto& product : products_) {
    char group = SymbolGroupConfig::GetInstance().
      GetGroupForRelativeNorm(mea, product->relative_norm());

    auto it = arbitrators_.find(group);
    if (it == arbitrators_.end()) {
      auto p = arbitrators_.insert(
        std::make_pair(
          group,
          std::make_unique<executor::TopicRecordArbitrator>(
            subscriber, arbitration_config.GetWindowSize(mea, group))));
      it = p.first;
    }

    auto topics = coin2::flow::TopicStringConfig::GetInstance()
      .ForProduct(mea, *product, workers, group);
    for (auto& topic : topics) {
      auto p = all_topics.insert(topic);
      if (p.second) {
        dist->Subscribe(topic, it->second.get());
      }
    }
  }
  return all_topics;
}

std::set<std::string> TopicRegister::RegisterFlowTopicsArb(
    const MarketExchangeApi& mea,
    const std::string& workers,
    const std::vector<std::unique_ptr<symbology::IProduct>>& products_,
    executor::TopicRecordSubscriber* subscriber,
    executor::TopicRecordDistributor* dist,
    const ArbitrationConfigMap& arbitration_config,
    const std::string& recipe) {
  std::set<std::string> all_topics;
  // interate symbols
  for (const auto& product : products_) {
    char group = SymbolGroupConfig::GetInstance().
      GetGroupForRelativeNorm(mea, product->relative_norm());

    auto it = archive_arbitrators_.find({group, recipe});
    if (it == archive_arbitrators_.end()) {
      auto p = archive_arbitrators_.insert(
        std::make_pair(
          std::make_pair(group, recipe),
          std::make_unique<executor::TopicRecordArbitrator>(
            subscriber, arbitration_config.GetWindowSize(mea, group))));
      it = p.first;
    }

    auto topics = coin2::flow::TopicStringConfig::GetInstance()
      .ForProduct(mea, *product, workers, group);
    for (auto& topic : topics) {
      auto p = all_topics.insert(topic);
      if (p.second) {
        dist->Subscribe(topic, it->second.get());
      }
    }
  }
  return all_topics;
}

std::set<std::string> TopicRegister::RegisterFlowTopics(
    const MarketExchangeApi& mea,
    const std::string& workers,
    const std::vector<std::unique_ptr<symbology::IProduct>>& products_,
    executor::TopicRecordSubscriber* subscriber,
    executor::TopicRecordDistributor* dist) {
  std::set<std::string> all_topics;
  // interate symbols
  for (const auto& product : products_) {
    char group = SymbolGroupConfig::GetInstance().
      GetGroupForRelativeNorm(mea, product->relative_norm());
    auto topics = coin2::flow::TopicStringConfig::GetInstance()
      .ForProduct(mea, *product, workers, group);
    for (auto& topic : topics) {
      auto p = all_topics.insert(topic);
      if (p.second) {
        dist->Subscribe(topic, subscriber);
      }
    }
  }
  return all_topics;
}

void TopicRegister::RegisterTopic(
    const std::string& topic,
    executor::TopicRecordSubscriber* subscriber,
    executor::TopicRecordDistributor* dist) {
  dist->Subscribe(topic, subscriber);
}

}  // namespace coin2::exchange::base::feed
